package cn.mesmile.storage.util;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.StrUtil;
import org.springframework.web.multipart.MultipartFile;

import java.io.*;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.StringJoiner;
import java.util.concurrent.*;

/**
 * @author zb
 * @date 2021/12/25 17:58
 * @Description 多线程断点续传工具
 */
public class TransportFileUtils {


    /**
     *  传输文件
     * @param src  源文件
     * @param desc 目标文件
     * @param threadNumber
     * @throws Exception
     */
    public static void transportUtils(File src, File desc, int threadNumber) throws Exception{
        // 获取绝对路径
        transportUtils(src.getCanonicalPath(), desc.getCanonicalPath(),5);
    }

    /**
     *  传输文件
     * @param src  源文件路径
     * @param desc 目标文件路径
     * @throws Exception
     */
    public static void transportUtils(String src, String desc) throws Exception{
        // 获取绝对路径
        transportUtils(src, desc,5);
    }


    /**
     *  传输文件
     * @param src  源文件路径
     * @param desc 目标文件路径
     * @param threadNumber 线程数
     * @throws Exception
     */
    public static void transportUtils(String src, String desc, int threadNumber) throws Exception{
        File file = FileUtil.newFile(src);
        // 文件总的大小
        long length = file.length();
        // 每个线程读取大小 向上取整
        final int part = (int)Math.ceil( length / threadNumber);

        List<Future<Object>> futures = new ArrayList<>();
        ExecutorService executorService = Executors.newFixedThreadPool(threadNumber);
        final Map<Integer, Integer> map = new ConcurrentHashMap<>(threadNumber);

        String canonicalPath = file.getCanonicalPath();
        String logPath = canonicalPath +".log" ;

        File logFile = new File(logPath);
        String[] data = null;
        if (logFile.exists()) {
            BufferedReader bufferedReader = new BufferedReader(new FileReader(logPath));
            String s = bufferedReader.readLine();
            if (StrUtil.isNotBlank(s)){
                data = s.split(",");
            }
            bufferedReader.close();
        }
        final String[] finalData = data;

        for (int i = 0; i < threadNumber; i++) {
            final int k = i;
            Future<Object> submit = executorService.submit(new Callable<Object>() {
                @Override
                public Object call() throws Exception {
                    // 自动关闭流
                    try (
                            RandomAccessFile in = new RandomAccessFile(file, "r");
                            RandomAccessFile out = new RandomAccessFile(desc, "rw");
                            RandomAccessFile log = new RandomAccessFile(logPath, "rw");
                    ) {
                        // 从什么时候开始写
                        in.seek(finalData == null ? k * part : Long.parseLong(finalData[k]));
                        out.seek(finalData == null ? k * part : Long.parseLong(finalData[k]));
                        byte[] bytes = new byte[1024 * 8];
                        int len = 0;
                        int totalSize = 0;
                        while (true) {
                            // 读
                            len = in.read(bytes);
                            if (len == -1) {
                                break;
                            }
                            // 总的大小
                            totalSize += len;

                            // key 为线程号  value为当前线程读取的位置
                            int currentValue = totalSize + (finalData == null ? k * part : Integer.parseInt(finalData[k]));
                            map.put(k, currentValue);

                            // 写
                            out.write(bytes, 0, len);
                            // 直接覆盖
                            log.seek(0);
                            StringJoiner stringJoiner = new StringJoiner(",");
                            map.forEach((key, value) -> stringJoiner.add(String.valueOf(value)));
                            log.write(stringJoiner.toString().getBytes(StandardCharsets.UTF_8));
                            // 若是断点续传，则需要大于下个线程的值
                            if (currentValue >= ((k + 1) * part)) {
                                break;
                            }
                        }
                        return Thread.currentThread().getId();
                    }
                }
            });
            futures.add(submit);
        }
        // 阻塞线程
        for (Future<Object> future : futures) {
            Object o = future.get();
        }
        // 最后删除日志文件
        logFile.delete();
    }



//    /**
//     *  传输文件
//     * @param src  源文件路径
//     * @param desc 目标文件路径
//     * @param threadNumber 线程数
//     * @throws Exception
//     */
//    public static void transportUtils(MultipartFile src, String desc, int threadNumber) throws Exception{
//        File descFile = FileUtil.file(desc);
//
//        InputStream inputStream = src.getInputStream();
//
//        // 文件总的大小
//        long length = src.getSize();
//        // 每个线程读取大小 向上取整
//        final int part = (int)Math.ceil( length / threadNumber);
//
//        List<Future<Object>> futures = new ArrayList<>();
//        ExecutorService executorService = Executors.newFixedThreadPool(threadNumber);
//        final Map<Integer, Integer> map = new ConcurrentHashMap<>(threadNumber);
//
//        String canonicalPath = descFile.getCanonicalPath();
//        String logPath = canonicalPath +".log" ;
//
//        File logFile = new File(logPath);
//        String[] data = null;
//        if (logFile.exists()) {
//            BufferedReader bufferedReader = new BufferedReader(new FileReader(logPath));
//            String s = bufferedReader.readLine();
//            if (StrUtil.isNotBlank(s)){
//                data = s.split(",");
//            }
//            bufferedReader.close();
//        }
//        final String[] finalData = data;
//
//        for (int i = 0; i < threadNumber; i++) {
//            final int k = i;
//            Future<Object> submit = executorService.submit(new Callable<Object>() {
//                @Override
//                public Object call() throws Exception {
//                    // 自动关闭流
//                    try (
//                            RandomAccessFile in = new RandomAccessFile(, "r");
//                            RandomAccessFile out = new RandomAccessFile(desc, "rw");
//                            RandomAccessFile log = new RandomAccessFile(logPath, "rw");
//                    ) {
//                        // 从什么时候开始写
//                        in.seek(finalData == null ? k * part : Long.parseLong(finalData[k]));
//                        out.seek(finalData == null ? k * part : Long.parseLong(finalData[k]));
//                        byte[] bytes = new byte[1024 * 8];
//                        int len = 0;
//                        int totalSize = 0;
//                        while (true) {
//                            // 读
//                            len = in.read(bytes);
//                            if (len == -1) {
//                                break;
//                            }
//                            // 总的大小
//                            totalSize += len;
//
//                            // key 为线程号  value为当前线程读取的位置
//                            int currentValue = totalSize + (finalData == null ? k * part : Integer.parseInt(finalData[k]));
//                            map.put(k, currentValue);
//
//                            // 写
//                            out.write(bytes, 0, len);
//                            // 直接覆盖
//                            log.seek(0);
//                            StringJoiner stringJoiner = new StringJoiner(",");
//                            map.forEach((key, value) -> stringJoiner.add(String.valueOf(value)));
//                            log.write(stringJoiner.toString().getBytes(StandardCharsets.UTF_8));
//                            // 若是断点续传，则需要大于下个线程的值
//                            if (currentValue >= ((k + 1) * part)) {
//                                break;
//                            }
//                        }
//                        return Thread.currentThread().getId();
//                    }
//                }
//            });
//            futures.add(submit);
//        }
//        // 阻塞线程
//        for (Future<Object> future : futures) {
//            Object o = future.get();
//        }
//        // 最后删除日志文件
//        logFile.delete();
//    }

}
